package org.apache.hadoop.hbase.util;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableDescriptors;
import org.apache.hadoop.hbase.TableInfoMissingException;

public class FSTableDescriptors
  implements TableDescriptors
{
  private static final Log LOG = LogFactory.getLog(FSTableDescriptors.class);
  private final FileSystem fs;
  private final Path rootdir;
  private final boolean fsreadonly;
  long cachehits = 0L;
  long invocations = 0L;
  public static final String TABLEINFO_NAME = ".tableinfo";
  private final Map<String, TableDescriptorModtime> cache = new ConcurrentHashMap();
  static final int WIDTH_OF_SEQUENCE_ID = 10;
  private static final Pattern SUFFIX = Pattern.compile(".tableinfo(\\.([0-9]{10}))?$");

  public FSTableDescriptors(FileSystem fs, Path rootdir)
  {
    this(fs, rootdir, false);
  }

  public FSTableDescriptors(FileSystem fs, Path rootdir, boolean fsreadOnly)
  {
    this.fs = fs;
    this.rootdir = rootdir;
    this.fsreadonly = fsreadOnly;
  }

  public HTableDescriptor get(byte[] tablename)
    throws IOException
  {
    return get(Bytes.toString(tablename));
  }

  public HTableDescriptor get(String tablename)
    throws IOException
  {
    this.invocations += 1L;
    if (HTableDescriptor.ROOT_TABLEDESC.getNameAsString().equals(tablename)) {
      this.cachehits += 1L;
      return HTableDescriptor.ROOT_TABLEDESC;
    }
    if (HTableDescriptor.META_TABLEDESC.getNameAsString().equals(tablename)) {
      this.cachehits += 1L;
      return HTableDescriptor.META_TABLEDESC;
    }

    if (HConstants.HBASE_NON_USER_TABLE_DIRS.contains(tablename)) {
      throw new IOException("No descriptor found for table = " + tablename);
    }

    TableDescriptorModtime cachedtdm = (TableDescriptorModtime)this.cache.get(tablename);

    if (cachedtdm != null)
    {
    	LOG.info("jzhd modify : cachehits="+this.cachehits);
      //if (getTableInfoModtime(this.fs, this.rootdir, tablename) <= cachedtdm.getModtime()) {
        this.cachehits += 1L;
        return cachedtdm.getTableDescriptor();
      //}
    }

    TableDescriptorModtime tdmt = null;
    try {
      tdmt = getTableDescriptorModtime(this.fs, this.rootdir, tablename);
    } catch (NullPointerException e) {
      LOG.debug("Exception during readTableDecriptor. Current table name = " + 
        tablename, e);
    } catch (IOException ioe) {
      LOG.debug("Exception during readTableDecriptor. Current table name = " + 
        tablename, ioe);
    }

    if (tdmt == null)
      LOG.warn("The following folder is in HBase's root directory and doesn't contain a table descriptor, do consider deleting it: " + 
        tablename);
    else {
      this.cache.put(tablename, tdmt);
    }
    return tdmt == null ? null : tdmt.getTableDescriptor();
  }

  public Map<String, HTableDescriptor> getAll()
    throws IOException
  {
    Map htds = new TreeMap();
    List<Path> tableDirs = FSUtils.getTableDirs(this.fs, this.rootdir);
    for (Path d : tableDirs) {
      HTableDescriptor htd = null;
      try
      {
        htd = get(d.getName());
      }
      catch (FileNotFoundException fnfe) {
        LOG.warn("Trouble retrieving htd", fnfe);
      }
      if (htd != null)
        htds.put(d.getName(), htd);
    }
    return htds;
  }

  public void add(HTableDescriptor htd) throws IOException
  {
    if (Bytes.equals(HConstants.ROOT_TABLE_NAME, htd.getName())) {
      throw new NotImplementedException();
    }
    if (Bytes.equals(HConstants.META_TABLE_NAME, htd.getName())) {
      throw new NotImplementedException();
    }
    if (HConstants.HBASE_NON_USER_TABLE_DIRS.contains(htd.getNameAsString())) {
      throw new NotImplementedException();
    }
    if (!this.fsreadonly) updateHTableDescriptor(this.fs, this.rootdir, htd);
    long modtime = getTableInfoModtime(this.fs, this.rootdir, htd.getNameAsString());
    this.cache.put(htd.getNameAsString(), new TableDescriptorModtime(modtime, htd));
  }

  public HTableDescriptor remove(String tablename)
    throws IOException
  {
    if (!this.fsreadonly) {
      Path tabledir = FSUtils.getTablePath(this.rootdir, tablename);
      if ((this.fs.exists(tabledir)) && 
        (!this.fs.delete(tabledir, true))) {
        throw new IOException("Failed delete of " + tabledir.toString());
      }
    }

    TableDescriptorModtime tdm = (TableDescriptorModtime)this.cache.remove(tablename);
    return tdm == null ? null : tdm.getTableDescriptor();
  }

  public static boolean isTableInfoExists(FileSystem fs, Path rootdir, String tableName)
    throws IOException
  {
    FileStatus status = getTableInfoPath(fs, rootdir, tableName);
    return status == null ? false : fs.exists(status.getPath());
  }

  private static FileStatus getTableInfoPath(FileSystem fs, Path rootdir, String tableName)
    throws IOException
  {
    Path tabledir = FSUtils.getTablePath(rootdir, tableName);
    return getTableInfoPath(fs, tabledir);
  }

  public static FileStatus getTableInfoPath(FileSystem fs, Path tabledir)
    throws IOException
  {
    FileStatus[] status = FSUtils.listStatus(fs, tabledir, new PathFilter()
    {
      public boolean accept(Path p)
      {
        return p.getName().startsWith(".tableinfo");
      }
    });
    if ((status == null) || (status.length < 1)) return null;
    Arrays.sort(status, new FileStatusFileNameComparator());
    if (status.length > 1)
    {
      for (int i = 1; i < status.length; i++) {
        Path p = status[i].getPath();

        if (!fs.delete(p, false))
          LOG.warn("Failed cleanup of " + status);
        else {
          LOG.debug("Cleaned up old tableinfo file " + p);
        }
      }
    }
    return status[0];
  }

  static String formatTableInfoSequenceId(int number)
  {
    byte[] b = new byte[10];
    int d = Math.abs(number);
    for (int i = b.length - 1; i >= 0; i--) {
      b[i] = (byte)(d % 10 + 48);
      d /= 10;
    }
    return Bytes.toString(b);
  }

  static int getTableInfoSequenceid(Path p)
  {
    if (p == null) return 0;
    Matcher m = SUFFIX.matcher(p.getName());
    if (!m.matches()) throw new IllegalArgumentException(p.toString());
    String suffix = m.group(2);
    if ((suffix == null) || (suffix.length() <= 0)) return 0;
    return Integer.parseInt(m.group(2));
  }

  static Path getTableInfoFileName(Path tabledir, int sequenceid)
  {
    return new Path(tabledir, 
      ".tableinfo." + formatTableInfoSequenceId(sequenceid));
  }

  static long getTableInfoModtime(FileSystem fs, Path rootdir, String tableName)
    throws IOException
  {
    FileStatus status = getTableInfoPath(fs, rootdir, tableName);
    return status == null ? 0L : status.getModificationTime();
  }

  public static HTableDescriptor getTableDescriptor(FileSystem fs, Path hbaseRootDir, byte[] tableName)
    throws IOException
  {
    HTableDescriptor htd = null;
    try {
      TableDescriptorModtime tdmt = 
        getTableDescriptorModtime(fs, hbaseRootDir, Bytes.toString(tableName));
      htd = tdmt == null ? null : tdmt.getTableDescriptor();
    } catch (NullPointerException e) {
      LOG.debug("Exception during readTableDecriptor. Current table name = " + 
        Bytes.toString(tableName), e);
    }
    return htd;
  }

  static HTableDescriptor getTableDescriptor(FileSystem fs, Path hbaseRootDir, String tableName) throws NullPointerException, IOException
  {
    TableDescriptorModtime tdmt = getTableDescriptorModtime(fs, hbaseRootDir, tableName);
    return tdmt == null ? null : tdmt.getTableDescriptor();
  }

  static TableDescriptorModtime getTableDescriptorModtime(FileSystem fs, Path hbaseRootDir, String tableName)
    throws NullPointerException, IOException
  {
    if ((Bytes.compareTo(Bytes.toBytes(tableName), HConstants.ROOT_TABLE_NAME) == 0) || 
      (Bytes.compareTo(Bytes.toBytes(tableName), HConstants.META_TABLE_NAME) == 0)) {
      return null;
    }
    return getTableDescriptorModtime(fs, FSUtils.getTablePath(hbaseRootDir, tableName));
  }

  static TableDescriptorModtime getTableDescriptorModtime(FileSystem fs, Path tableDir) throws NullPointerException, IOException
  {
    if (tableDir == null) throw new NullPointerException();
    FileStatus status = getTableInfoPath(fs, tableDir);
    if (status == null) {
      throw new TableInfoMissingException("No .tableinfo file under " + 
        tableDir.toUri());
    }
    FSDataInputStream fsDataInputStream = fs.open(status.getPath());
    HTableDescriptor hTableDescriptor = null;
    try {
      hTableDescriptor = new HTableDescriptor();
      hTableDescriptor.readFields(fsDataInputStream);
    } finally {
      fsDataInputStream.close();
    }
    return new TableDescriptorModtime(status.getModificationTime(), hTableDescriptor);
  }

  public static HTableDescriptor getTableDescriptor(FileSystem fs, Path tableDir) throws IOException, NullPointerException
  {
    TableDescriptorModtime tdmt = getTableDescriptorModtime(fs, tableDir);
    return tdmt == null ? null : tdmt.getTableDescriptor();
  }

  static Path updateHTableDescriptor(FileSystem fs, Path rootdir, HTableDescriptor hTableDescriptor)
    throws IOException
  {
    Path tableDir = FSUtils.getTablePath(rootdir, hTableDescriptor.getName());
    Path p = writeTableDescriptor(fs, hTableDescriptor, tableDir, 
      getTableInfoPath(fs, tableDir));
    if (p == null) throw new IOException("Failed update");
    LOG.info("Updated tableinfo=" + p);
    return p;
  }

  public static void deleteTableDescriptorIfExists(String tableName, Configuration conf)
    throws IOException
  {
    FileSystem fs = FSUtils.getCurrentFileSystem(conf);
    FileStatus status = getTableInfoPath(fs, FSUtils.getRootDir(conf), tableName);

    if ((status != null) && (fs.exists(status.getPath())))
      FSUtils.deleteDirectory(fs, status.getPath());
  }

  private static Path writeTableDescriptor(FileSystem fs, HTableDescriptor hTableDescriptor, Path tableDir, FileStatus status)
    throws IOException
  {
    Path tmpTableDir = new Path(tableDir, ".tmp");

    int currentSequenceid = 
      status == null ? 0 : getTableInfoSequenceid(status.getPath());
    int sequenceid = currentSequenceid;

    int retries = 10;
    int retrymax = currentSequenceid + retries;
    Path tableInfoPath = null;
    do {
      sequenceid++;
      Path p = getTableInfoFileName(tmpTableDir, sequenceid);
      if (fs.exists(p)) {
        LOG.debug(p + " exists; retrying up to " + retries + " times");
      }
      else {
        try {
          writeHTD(fs, p, hTableDescriptor);
          tableInfoPath = getTableInfoFileName(tableDir, sequenceid);
          if (!fs.rename(p, tableInfoPath))
            throw new IOException("Failed rename of " + p + " to " + tableInfoPath);
        }
        catch (IOException ioe)
        {
          LOG.debug("Failed write and/or rename; retrying", ioe);
          if (!FSUtils.deleteDirectory(fs, p)) {
            LOG.warn("Failed cleanup of " + p);
          }
          tableInfoPath = null;
        }

        if ((status == null) || 
          (FSUtils.deleteDirectory(fs, status.getPath()))) break;
        LOG.warn("Failed delete of " + status.getPath() + "; continuing");

        break;
      }
    }
    while (
      sequenceid < retrymax);
    return tableInfoPath;
  }

  private static void writeHTD(FileSystem fs, Path p, HTableDescriptor htd)
    throws IOException
  {
    FSDataOutputStream out = fs.create(p, false);
    try {
      htd.write(out);
      out.write(10);
      out.write(10);
      out.write(Bytes.toBytes(htd.toString()));
    } finally {
      out.close();
    }
  }

  public static boolean createTableDescriptor(HTableDescriptor htableDescriptor, Configuration conf)
    throws IOException
  {
    return createTableDescriptor(htableDescriptor, conf, false);
  }

  static boolean createTableDescriptor(HTableDescriptor htableDescriptor, Configuration conf, boolean forceCreation)
    throws IOException
  {
    FileSystem fs = FSUtils.getCurrentFileSystem(conf);
    return createTableDescriptor(fs, FSUtils.getRootDir(conf), htableDescriptor, 
      forceCreation);
  }

  public static boolean createTableDescriptor(FileSystem fs, Path rootdir, HTableDescriptor htableDescriptor)
    throws IOException
  {
    return createTableDescriptor(fs, rootdir, htableDescriptor, false);
  }

  public static boolean createTableDescriptor(FileSystem fs, Path rootdir, HTableDescriptor htableDescriptor, boolean forceCreation)
    throws IOException
  {
    Path tabledir = FSUtils.getTablePath(rootdir, htableDescriptor.getNameAsString());
    return createTableDescriptorForTableDirectory(fs, tabledir, htableDescriptor, forceCreation);
  }

  public static boolean createTableDescriptorForTableDirectory(FileSystem fs, Path tabledir, HTableDescriptor htableDescriptor, boolean forceCreation)
    throws IOException
  {
    FileStatus status = getTableInfoPath(fs, tabledir);
    if (status != null) {
      LOG.info("Current tableInfoPath = " + status.getPath());
      if ((!forceCreation) && 
        (fs.exists(status.getPath())) && (status.getLen() > 0L)) {
        LOG.info("TableInfo already exists.. Skipping creation");
        return false;
      }
    }

    Path p = writeTableDescriptor(fs, htableDescriptor, tabledir, status);
    return p != null;
  }

  static class FileStatusFileNameComparator
    implements Comparator<FileStatus>
  {
    public int compare(FileStatus left, FileStatus right)
    {
      return -left.compareTo(right);
    }
  }

  static class TableDescriptorModtime
  {
    private final HTableDescriptor descriptor;
    private final long modtime;

    TableDescriptorModtime(long modtime, HTableDescriptor htd)
    {
      this.descriptor = htd;
      this.modtime = modtime;
    }

    long getModtime() {
      return this.modtime;
    }

    HTableDescriptor getTableDescriptor() {
      return this.descriptor;
    }
  }
}